﻿using log4net;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using RabbitMQ.Client.Exceptions;
using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;

namespace RabbitMQShare
{
    public class RabbitMQUtil
    {
        // 保存队列key值，用来查询队列是否包含消息
        public static List<int> _keys=new List<int>();
        public static ConnectionFactory _connectionFactory;

        public static void InitFactory(RabbitMqModel mqModel)
        {
            if (_connectionFactory != null) return;
            _connectionFactory = new ConnectionFactory()
            {
                HostName = mqModel.Host,
                Port = mqModel.Port,
                UserName = mqModel.UserName,
                Password = mqModel.PassWord
            };
        }


        // 发送消息
        public static bool SendMsgToServer(object content,string queueName,string exchangeName,ILog log)
        {
            if (_connectionFactory == null)
            {
                // 发送消息必须先实例化工厂
                return false;
            }
            bool flag;
            _connectionFactory.RequestedHeartbeat= TimeSpan.FromSeconds(10);
            try
            {
                using (var connection = _connectionFactory.CreateConnection())
                {
                    //创建一个通道，这个就是Rabbit自己定义的规则了，如果自己写消息队列，这个就可以开脑洞设计了
                    //这里Rabbit的玩法就是一个通道channel下包含多个队列Queue
                    //http://t.csdn.cn/SqqQO
                    using (var channel = connection.CreateModel())
                    {
                        // queue: 队列的名称
                        // durable: 设置是否持久化, true表示队列为持久化, 持久化的队列会存盘, 在服务器重启的时候会保证不丢失相关信息
                        // exclusive: 设置是否排他, true表示队列为排他的, 如果一个队列被设置为排他队列, 该队列仅对首次声明它的连接可见, 并在连接断开时自动删除, (这里需要注意三点:1.排他队列是基于连接Connection可见的, 同一个连接的不同信道Channel是可以同时访问同一连接创建的排他队列;"首次"是指如果一个连接己经声明了一个排他队列，其他连接是不允许建立同名的排他队列的，这个与普通队列不同;即使该队列是持久化的，一旦连接关闭或者客户端退出，该排他队列都会被自动删除，这种队列适用于一个客户端同时发送和读取消息的应用场景)
                        //https://www.cnblogs.com/no-celery/p/13803419.html
                        //声明交换机exchang
                        channel.ExchangeDeclare(exchange: "myexchange",
                                                type: ExchangeType.Direct,
                                                durable: false,
                                                autoDelete: false,
                                                arguments: null);
                        channel.QueueDeclare(queueName, false, false, false,
                            arguments: new Dictionary<string, object>() {
                                  { "x-max-priority",10 }
                              }
                            );
                        channel.QueueBind(queue: queueName, exchange: "myexchange", "");
                        channel.ConfirmSelect();
                        //channel.ExchangeDeclare(exchangeName, ExchangeType.Direct, false, false, null);
                        //channel.QueueBind(queueName, exchangeName, "", null);
                        //var properties = channel.CreateBasicProperties();
                        //properties.DeliveryMode = 1; // 2为消息持久化，前提条件是队列也持久化，不然没有意义
                        // 序列化对象 发送消息
                        string message = JsonConvert.SerializeObject(content);
                        //channel.BasicPublish("", queueName, null, Encoding.UTF8.GetBytes(message));
                        var properties = channel.CreateBasicProperties();
                        properties.Persistent = true;
                        properties.Priority = 1;
                        //properties.MessageId = Guid.NewGuid().ToString("N");
                        //properties.Timestamp = new AmqpTimestamp(DateTimeOffset.UtcNow.ToUnixTimeMilliseconds());
                        //properties.Headers = new Dictionary<string, object>
                        //{
                        //    { "key", "value1"}
                        //};
                        var body = Encoding.UTF8.GetBytes(message);
                        channel.BasicPublish("myexchange", "",properties, body);
                        var isOk = channel.WaitForConfirms();
                        if (!isOk)
                        {
                            //throw new Exception("The message is not reached to the server!");
                            flag = false;
                        }
                        flag = true;
                        //channel.BasicReturn += (sender, ea) =>
                        //{
                        //    // 消息发送失败
                        //    flag = false;
                        //};
                    }
                }
                //_connectionFactory = null;
            }
            catch (Exception ex){
                log.Error(ex.Message, ex);
                flag = false;
            }
            return flag;
        }

        // 有优先等级的消息发送
        public static bool SendMsgToServerWithLevel(object content, string queueName, string exchangeName,ILog log)
        {
            if (_connectionFactory == null)
            {
                // 发送消息必须先实例化工厂
                return false;
            }
            bool flag;
            log.Info("发送消息");
            _connectionFactory.RequestedHeartbeat = TimeSpan.FromSeconds(10);
            try
            {
                using (var connection = _connectionFactory.CreateConnection())
                {
                    //创建一个通道，这个就是Rabbit自己定义的规则了，如果自己写消息队列，这个就可以开脑洞设计了
                    //这里Rabbit的玩法就是一个通道channel下包含多个队列Queue
                    //http://t.csdn.cn/SqqQO
                    using (var channel = connection.CreateModel())
                    {
                        //声明交换机exchang
                        channel.ExchangeDeclare(exchange: "myexchange",
                                                type: ExchangeType.Direct,
                                                durable: false,
                                                autoDelete: false,
                                                arguments: null);
                        // queue: 队列的名称
                        // durable: 设置是否持久化, true表示队列为持久化, 持久化的队列会存盘, 在服务器重启的时候会保证不丢失相关信息
                        // exclusive: 设置是否排他, true表示队列为排他的, 如果一个队列被设置为排他队列, 该队列仅对首次声明它的连接可见, 并在连接断开时自动删除, (这里需要注意三点:1.排他队列是基于连接Connection可见的, 同一个连接的不同信道Channel是可以同时访问同一连接创建的排他队列;"首次"是指如果一个连接己经声明了一个排他队列，其他连接是不允许建立同名的排他队列的，这个与普通队列不同;即使该队列是持久化的，一旦连接关闭或者客户端退出，该排他队列都会被自动删除，这种队列适用于一个客户端同时发送和读取消息的应用场景)
                        //https://www.cnblogs.com/no-celery/p/13803419.html
                        channel.QueueDeclare(queueName, false, false, false, 
                              arguments:new Dictionary<string, object>() {
                                  { "x-max-priority",10 }
                              }
                            );
                        channel.QueueBind(queue: queueName, exchange: "myexchange","");
                        channel.ConfirmSelect();
                        //channel.ExchangeDeclare(exchangeName, ExchangeType.Direct, false, false, null);
                        //channel.QueueBind(queueName, exchangeName, "", null);
                        //var properties = channel.CreateBasicProperties();
                        //properties.DeliveryMode = 1; // 2为消息持久化，前提条件是队列也持久化，不然没有意义
                        // 序列化对象 发送消息
                        string message = JsonConvert.SerializeObject(content);
                        //channel.BasicPublish("", queueName, null, Encoding.UTF8.GetBytes(message));
                        var properties = channel.CreateBasicProperties();
                        JObject obj = JsonConvert.DeserializeObject<JObject>(message);
                        if (!string.IsNullOrEmpty(obj["Remark"].ToString()) && obj["Remark"].ToString().Equals("vip"))
                        {
                            properties.Priority = 9;// 设置优先级
                        }
                        else
                        {
                            properties.Priority = 1;
                        }
                        properties.Persistent = true; // 
                        //properties.MessageId = Guid.NewGuid().ToString("N");
                        //properties.Timestamp = new AmqpTimestamp(DateTimeOffset.UtcNow.ToUnixTimeMilliseconds());
                        //properties.Headers = new Dictionary<string, object>
                        //{
                        //    { "key", "value1"}
                        //};
                        var body = Encoding.UTF8.GetBytes(message);
                        channel.BasicPublish("myexchange", "", properties,body);
                        var isOk = channel.WaitForConfirms();
                        if (!isOk)
                        {
                            //throw new Exception("The message is not reached to the server!");
                            flag = false;
                        }
                        flag = true;
                        //channel.BasicReturn += (sender, ea) =>
                        //{
                        //    // 消息发送失败
                        //    flag = false;
                        //};
                    }
                }
                //_connectionFactory = null;
            }
            catch (Exception ex)
            {
                log.Error(ex.Message, ex);
                flag = false;
            }
            return flag;
        }


        // 接收消息
        public static T GetMsgFromQueue<T>(string queueName,string exchangeName)
        {
            if (_connectionFactory == null)
                return default(T);
            using (var connection=_connectionFactory.CreateConnection())
            {
                using (var channel = connection.CreateModel())
                {
                    // 定义通道
                    channel.BasicQos(0, 1, false);
                    channel.QueueDeclare(queueName, false, false, false, null);
                    //channel.ExchangeDeclare(exchangeName, ExchangeType.Direct, false, false, null);
                    //channel.QueueBind(queueName, exchangeName, "", null);
                    var consumer = new EventingBasicConsumer(channel);//消费者
                    // 接收处理消息
                    T model = default(T);
                    consumer.Received += (model, ea) =>
                    {
                        string message = Encoding.UTF8.GetString(ea.Body.ToArray());//接收消息
                        // 反序列化
                        //model = JsonConvert.DeserializeObject<T>(message);
                        // 模拟处理数据
                        //Thread.Sleep(8000);
                        // 消息确认
                        channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false); // multiple 是否批量true：将一次性ack所有小于deliveryTag的消息；确认收到消息。
                    };
                    channel.BasicConsume(queueName, false, consumer);//消费消息  //autoAck true 才会删除确认的消息
                    _connectionFactory = null;
                    return model;
                }
            }
        }
        public static void GetMsgFromQueue2<T>(string queueName)
        {
            using (IConnection connection = _connectionFactory.CreateConnection())
            {
                using (IModel channel = connection.CreateModel())
                {
                    channel.BasicQos(0, 1, false);

                    //channel.ExchangeDeclare("Exchange", "direct");
                    channel.QueueDeclare(queueName, false, false, false);
                    //channel.QueueBind(queueName, "Exchange", "RouteA");
                    Console.WriteLine("Waiting for message");
                    var consumer = new EventingBasicConsumer(channel);
                    consumer.Received += (model, ea) =>
                    {
                        Console.WriteLine("开始处理数据..");
                        //The QueueA is always successful.
                        string message = Encoding.UTF8.GetString(ea.Body.ToArray());//接收消息
                        if (string.IsNullOrEmpty(message))
                        {
                            Console.WriteLine("数据接收为空");
                        }
                        else
                        {
                            // 落库保存数据
                            Console.WriteLine(message);
                        }
                        // 模拟处理任务，延时20秒
                        Thread.Sleep(20000);
                        try
                        {
                            channel.BasicAck(ea.DeliveryTag, false);
                        }
                        catch (AlreadyClosedException ex)
                        {

                        }
                    };
                    channel.BasicConsume("testQue1", false, consumer);
                    Console.WriteLine("Press [enter] to exit.");
                    Console.ReadLine();
                }
            }
        }
    }
}
